package com.sip.stream.rxbus;

import android.util.Log;
import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableEmitter;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.Scheduler;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.functions.Predicate;
import io.reactivex.processors.FlowableProcessor;
import io.reactivex.processors.PublishProcessor;
import io.reactivex.schedulers.Schedulers;

public final class RxBus {
    private final FlowableProcessor<Object> mBus;
    private final Consumer<Throwable> mOnError;

    public static abstract class Callback<T> {
        public abstract void onEvent(T t);
    }

    private RxBus() {
        this.mOnError = new Consumer<Throwable>() {
            @Override
            public void accept(Throwable th) {
                Utils.logE(th.toString());
                Log.w("RXBUS", "accept " + th.toString());
                th.printStackTrace();
            }
        };
        this.mBus = PublishProcessor.create().toSerialized();
    }

    public static RxBus getDefault() {
        return Holder.BUS;
    }

    public void postDelaySeconds(Object obj, String str, long j) {
        Utils.requireNonNull(obj, str);
        new TagMessage(obj, str);
    }

    public void post(Object obj, String str) {
        post(obj, str, false);
    }

    public void postSticky(Object obj) {
        post(obj, "", true);
    }

    public void postSticky(Object obj, String str) {
        post(obj, str, true);
    }

    private void post(Object obj, String str, boolean z) {
        Utils.requireNonNull(obj, str);
        TagMessage tagMessage = new TagMessage(obj, str);
        if (z) {
            CacheUtils.getInstance().addStickyEvent(obj, str);
        }
        this.mBus.onNext(tagMessage);
    }

    public void removeSticky(Object obj) {
        removeSticky(obj, "");
    }

    public void removeSticky(Object obj, String str) {
        Utils.requireNonNull(obj, str);
        CacheUtils.getInstance().removeStickyEvent(obj, str);
    }

    public <T> void subscribe(Object obj, Callback<T> callback) {
        subscribe(obj, "", false, null, callback);
    }

    public <T> void subscribe(Object obj, String str, Callback<T> callback) {
        subscribe(obj, str, false, null, callback);
    }

    public <T> void subscribe(Object obj, Scheduler scheduler, Callback<T> callback) {
        subscribe(obj, "", false, scheduler, callback);
    }

    public <T> void subscribe(Object obj, String str, Scheduler scheduler, Callback<T> callback) {
        subscribe(obj, str, false, scheduler, callback);
    }

    public <T> void subscribeSticky(Object obj, Callback<T> callback) {
        subscribe(obj, "", true, null, callback);
    }

    public <T> void subscribeSticky(Object obj, String str, Callback<T> callback) {
        subscribe(obj, str, true, null, callback);
    }

    public <T> void subscribeSticky(Object obj, Scheduler scheduler, Callback<T> callback) {
        subscribe(obj, "", true, scheduler, callback);
    }

    public <T> void subscribeSticky(Object obj, String str, Scheduler scheduler, Callback<T> callback) {
        subscribe(obj, str, true, scheduler, callback);
    }

    private <T> void subscribe(Object obj, String str, boolean z, Scheduler scheduler, final Callback<T> callback) {
        Utils.requireNonNull(obj, str, callback);
        final Class<T> typeClassFromParadigm = Utils.getTypeClassFromParadigm(callback);
        Consumer<T> consumer = new Consumer<T>() {
            @Override
            public void accept(T t) {
                callback.onEvent(t);
            }
        };
        if (z) {
            final TagMessage findStickyEvent = CacheUtils.getInstance().findStickyEvent(typeClassFromParadigm, str);
            if (findStickyEvent != null) {
                Flowable<T> create = Flowable.create(new FlowableOnSubscribe<T>() {
                    @Override
                    public void subscribe(FlowableEmitter<T> flowableEmitter) {
                        flowableEmitter.onNext(typeClassFromParadigm.cast(findStickyEvent.mEvent));
                    }
                }, BackpressureStrategy.LATEST);
                if (scheduler != null) {
                    create = create.observeOn(scheduler);
                }
                CacheUtils.getInstance().addDisposable(obj, FlowableUtils.subscribe(create, consumer, this.mOnError));
            } else {
                Utils.logW("sticky event is empty.");
            }
        }
        CacheUtils.getInstance().addDisposable(obj, FlowableUtils.subscribe(toFlowable(typeClassFromParadigm, str, scheduler), consumer, this.mOnError));
    }

    private <T> Flowable<T> toFlowable(final Class<T> cls, final String str, Scheduler scheduler) {
        Flowable<T> doOnError = this.mBus.ofType(TagMessage.class).filter(new Predicate<TagMessage>() {
            @Override
            public boolean test(TagMessage tagMessage) {
                return tagMessage.isSameType(cls, str);
            }
        }).map(new Function<TagMessage, Object>() {
            @Override
            public Object apply(TagMessage tagMessage) {
                return tagMessage.mEvent;
            }
        }).cast(cls).onBackpressureDrop(new Consumer<T>() {
            @Override
            public void accept(T t) throws Exception {
                Log.w("RXBUS", "onBackpressureDrop " + t.toString());
            }
        }).doOnError(new Consumer<Throwable>() {
            @Override
            public void accept(Throwable th) throws Exception {
                Log.w("RXBUS", "toFlowable throwable " + th.getMessage());
                th.printStackTrace();
            }
        });
        if (scheduler != null) {
            return doOnError.observeOn(scheduler, false, 256);
        }
        return doOnError.observeOn(Schedulers.io(), false, 256);
    }

    public void unregister(Object obj) {
        CacheUtils.getInstance().removeDisposables(obj);
    }

    private static class Holder {
        private static final RxBus BUS = new RxBus();

        private Holder() {
        }
    }
}
